package io.xxx.xbutler.core.select.receiver;

import cn.idev.excel.FastExcel;
import cn.idev.excel.context.AnalysisContext;
import cn.idev.excel.read.listener.ReadListener;
import cn.idev.excel.support.ExcelTypeEnum;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.aliyun.oss.OSS;
import io.xxx.xbutler.domain.Contact;
import io.xxx.xbutler.domain.Room;
import io.xxx.xbutler.domain.SelectMode;
import io.xxx.xbutler.domain.task.Receiver;
import io.xxx.xbutler.domain.task.ReceiverType;
import io.xxx.xbutler.domain.task.Task;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;

import javax.annotation.CheckReturnValue;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

@Slf4j
@Component
public class FileReceiverSelector extends AbstractReceiverSelector {

    @Resource
    private OSS oss;

    @Value("${task.select.file.bucket:leo}")
    private String bucket;

    @Value("${task.select.file.batch-size:1000}")
    private Integer batchSize;

    @Override
    public Flux<Receiver> select(Task task) {
        JSONObject selectContent = JSON.parseObject(task.getSelectContent());
        String fileUrl = selectContent.getString("url");
        ReceiverType receiverType = task.getReceiverType();
        InputStream inputStream = getInputStream(task.getId(), fileUrl);
        if (inputStream == null) {
            return Flux.empty();
        }
        return Flux.create(emitter -> {
            try {
                List<String> storeCodes = new ArrayList<>(batchSize);
                FastExcel.read(inputStream, new ReadListener<Map<Integer, String>>() {
                            @Override
                            public void invoke(Map<Integer, String> data, AnalysisContext context) {
                                String storeCode = data.get(0);
                                if (!StringUtils.hasText(storeCode)) {
                                    return;
                                }
                                storeCodes.add(storeCode);

                                if (storeCodes.size() == batchSize) {
                                    emit(receiverType, storeCodes, emitter);
                                    storeCodes.clear();
                                }
                            }

                            @Override
                            public void doAfterAllAnalysed(AnalysisContext context) {
                                if (storeCodes.isEmpty()) {
                                    return;
                                }
                                emit(receiverType, storeCodes, emitter);
                                storeCodes.clear();
                            }
                        })
                        .excelType(ExcelTypeEnum.XLSX)
                        .doReadAll();
                emitter.complete();
            } catch (Exception e) {
                log.error("读取文件异常[taskId:{}, fileUrl:{}]", task.getId(), fileUrl, e);
                emitter.error(e);
            }
        });
    }

    private void emit(ReceiverType receiverType, List<String> storeCodes, FluxSink<Receiver> emitter) {
        switch (receiverType) {
            case CONTACT: {
                List<Contact> contacts = contactMapper.selectListByStoreCodes(storeCodes);
                for (Contact contact : contacts) {
                    Receiver receiver = buildReceiver(contact);
                    emitter.next(receiver);
                }
            }
            case ROOM: {
                List<Room> rooms = roomMapper.selectListByStoreCodes(storeCodes);
                for (Room room : rooms) {
                    Receiver receiver = buildReceiver(room);
                    emitter.next(receiver);
                }
            }
        }
    }

    @CheckReturnValue
    protected InputStream getInputStream(Long taskId, String fileUrl) {
        if (!StringUtils.hasText(fileUrl)) {
            log.error("文件地址不存在[taskId:{}, fileUrl:{}]", taskId, fileUrl);
            return null;
        }
        return oss.getObject(bucket, fileUrl).getObjectContent();
    }

    @Override
    public boolean supports(SelectMode selectMode) {
        return selectMode == SelectMode.FILE;
    }
}
